package com.hdw.job.batch.mapper;

import com.hdw.common.base.BaseException;
import com.hdw.job.constant.JobConstant;
import com.hdw.job.batch.element.*;
import org.apache.commons.lang3.StringUtils;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.stereotype.Component;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Types;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Component
@StepScope
public class JDBCWriterRowMapper implements RowMapper {
    @Value("#{jobParameters[characterSetSou]}")
    String characterSetSou;
    @Value("#{jobParameters[jobType]}")
    String jobType;

    private static final byte[] EMPTY_CHAR_ARRAY = new byte[0];
    private ResultSetMetaData metaData;
    private int columnNumber = -1;
    //public static final String ADD_COLUMN_DT_SYS_CRE = "DT_SYS_CRE";
    //private Date dtSysCre = new Date();
    //public static final String ADD_COLUMN_DT_SYS_MODI = "DT_SYS_MODI";
    //private Date dtSysModi = new Date();
    public static final String ADD_COLUMN_EU_TP = "EU_TP";


    @Override
    public Map<String, Column> mapRow(ResultSet rs, int rowNum) throws SQLException {
        if (metaData == null) metaData = rs.getMetaData();
        if (columnNumber <= 0) columnNumber = metaData.getColumnCount();

        Map<String, Column> customerCredit = buildRecord(rs, metaData, columnNumber, characterSetSou);


        customerCredit = addDefaultColumn(customerCredit);

        return customerCredit;
    }

    private Map addDefaultColumn(Map customerCredit) {
        if (JobConstant.JOB_TYPE_DIM.equalsIgnoreCase(jobType)) {
            customerCredit.put(ADD_COLUMN_EU_TP, new LongColumn(1));
        }

        return customerCredit;
    }

    protected Map<String, Column> buildRecord(ResultSet rs, ResultSetMetaData metaData, int columnNumber, String mandatoryEncoding) {
        Map<String, Column> record = new HashMap();

        try {
            for (int i = 1; i <= columnNumber; i++) {
                String columnName = metaData.getColumnName(i);

                switch (metaData.getColumnType(i)) {

                    case Types.CHAR:
                    case Types.NCHAR:
                    case Types.VARCHAR:
                    case Types.LONGVARCHAR:
                    case Types.NVARCHAR:
                    case Types.LONGNVARCHAR:
                        String rawData;
                        if (StringUtils.isBlank(mandatoryEncoding)) {
                            rawData = rs.getString(i);
                        } else {
                            rawData = new String((rs.getBytes(i) == null ? EMPTY_CHAR_ARRAY :
                                    rs.getBytes(i)), mandatoryEncoding);
                        }
                        record.put(columnName, new StringColumn(rawData));
                        break;

                    case Types.CLOB:
                    case Types.NCLOB:
                        record.put(columnName, new StringColumn(rs.getString(i)));
                        break;

                    case Types.SMALLINT:
                    case Types.TINYINT:
                    case Types.INTEGER:
                    case Types.BIGINT:
                        record.put(columnName, new LongColumn(rs.getString(i)));
                        break;

                    case Types.NUMERIC:
                    case Types.DECIMAL:
                        record.put(columnName, new DoubleColumn(rs.getString(i)));
                        break;

                    case Types.FLOAT:
                    case Types.REAL:
                    case Types.DOUBLE:
                        record.put(columnName, new DoubleColumn(rs.getString(i)));
                        break;

                    case Types.TIME:
                        record.put(columnName, new DateColumn(rs.getTime(i)));
                        break;

                    // for mysql bug, see http://bugs.mysql.com/bug.php?id=35115
                    case Types.DATE:
                        if (metaData.getColumnTypeName(i).equalsIgnoreCase("year")) {
                            record.put(columnName, new LongColumn(rs.getInt(i)));
                        } else {
                            record.put(columnName, new DateColumn(rs.getDate(i)));
                        }
                        break;

                    case Types.TIMESTAMP:
                        record.put(columnName, new DateColumn(rs.getTimestamp(i)));
                        break;

                    case Types.BINARY:
                    case Types.VARBINARY:
                    case Types.BLOB:
                    case Types.LONGVARBINARY:
                        record.put(columnName, new BytesColumn(rs.getBytes(i)));
                        break;

                    // warn: bit(1) -> Types.BIT 可使用BoolColumn
                    // warn: bit(>1) -> Types.VARBINARY 可使用BytesColumn
                    case Types.BOOLEAN:
                    case Types.BIT:
                        record.put(columnName, new BoolColumn(rs.getBoolean(i)));
                        break;

                    case Types.NULL:
                        String stringData = null;
                        if (rs.getObject(i) != null) {
                            stringData = rs.getObject(i).toString();
                        }
                        record.put(columnName, new StringColumn(stringData));
                        break;

                    default:
                        throw new BaseException(String.format(
                                "您的配置文件中的列配置信息有误. 不支持数据库读取这种字段类型. 字段名:[%s], 字段名称:[%s], 字段Java类型:[%s]. 请尝试使用数据库函数将其转换支持的类型 或者不同步该字段 .",
                                metaData.getColumnName(i),
                                metaData.getColumnType(i),
                                metaData.getColumnClassName(i)));
                }
            }
        } catch (Exception e) {

        }
        return record;
    }
}

